package cn.doitedu.demo7;

import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.*;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.nio.ByteBuffer;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 * 实现： 可以动态注入、下线、更新规则
 * 当然，前提是，这些规则，都是系统已经内置的 规则模型下的规则
 **/
@Slf4j
public class MainEntry2_OperatorState {

    //private static final Logger log = LoggerFactory.getLogger(MainEntry.class);

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setStateBackend(new HashMapStateBackend());


        // 构造cdc连接器，实时抓取 规则元数据表中的数据变更
        MySqlSource<String> ruleMetaCdcSource = MySqlSource.<String>builder()
                .hostname("doitedu")
                .port(3306)
                .databaseList("doit46") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("doit46.rule_meta") // set captured table
                .username("root")
                .password("root")
                .deserializer(new JsonDebeziumDeserializationSchema()) // converts SourceRecord to JSON String
                .build();


        // 解析cdc抓取的规则元数据变更
        DataStreamSource<String> cdcStream = env.fromSource(ruleMetaCdcSource, WatermarkStrategy.noWatermarks(), "cdc");

        // 对抓取到的ruleMeta变更原始json数据，进行解析、提取
        SingleOutputStreamOperator<RuleMeta> ruleMetaBeans = cdcStream.map(json -> {

            JSONObject jsonObject = JSON.parseObject(json);
            String op = jsonObject.getString("op");

            JSONObject dataObj;
            if (op.equals("d")) {
                dataObj = jsonObject.getJSONObject("before");
            } else {
                dataObj = jsonObject.getJSONObject("after");
            }

            byte[] bytes = dataObj.getBytes("pre_select_users");

            String jsonString = dataObj.toJSONString();
            RuleMeta ruleMeta = JSON.parseObject(jsonString, RuleMeta.class);

            // 填充op字段
            ruleMeta.setOp(op);

            // 反序列化 bitmap的字节数组
            byte[] preSelectUsersBytes = ruleMeta.getPre_select_users();
            Roaring64Bitmap bitmap = Roaring64Bitmap.bitmapOf();

            if (preSelectUsersBytes != null) {
                bitmap.deserialize(ByteBuffer.wrap(preSelectUsersBytes));
                log.info("反序列化得到的bitmap是: {}", bitmap);
            }
            // 填充bitmap字段
            ruleMeta.setPreSelectBitmap(bitmap);

            return ruleMeta;
        });


        // 将规则元数据流，进行广播
        MapStateDescriptor<String, String> desc = new MapStateDescriptor<>("bc_state", String.class, String.class);
        BroadcastStream<RuleMeta> ruleMetaBroadcastStream = ruleMetaBeans.broadcast(desc);


        // 构建kafka source来读取数据
        KafkaSource<String> source = KafkaSource
                .<String>builder()
                .setBootstrapServers("doitedu:9092")
                .setTopics("user-action-log")
                .setStartingOffsets(OffsetsInitializer.latest())
                .setGroupId("doitedu-ww")
                .setClientIdPrefix("doitedu-cc")
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();


        DataStreamSource<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "log-source");

        // 将source读取到的json数据，转换成javaBean数据
        DataStream<UserAction> mapped = stream.map(json -> JSON.parseObject(json, UserAction.class));

        // 按用户分区
        KeyedStream<UserAction, Long> keyedStream = mapped.keyBy(UserAction::getUser_id);


        // 将用户行为流 ， 连接   规则元数据广播流
        keyedStream.connect(ruleMetaBroadcastStream)
                .process(new CoreProcessFunction_OperatorState()).print();


        env.execute();


    }

}
